package com.aliyun.commons.service.impl;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;

import javax.annotation.PostConstruct;

import com.aliyun.commons.service.SendMessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.ProducerBuilder;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.apis.producer.Transaction;
import org.apache.rocketmq.client.apis.producer.TransactionChecker;

/**
 * 发送消息服务实现
 *
 * 
 * @date 2024/11/30
 */
@Slf4j
public class SendMessageServiceImpl implements SendMessageService {

    private static final ClientServiceProvider provider = ClientServiceProvider.loadService();
    private static final ExecutorService sendCallbackExecutor = Executors.newFixedThreadPool(1);
    private Producer producer;

    private String accessKey;
    private String accessSecret;
    private String endpoint;
    private List<String> topics;
    private TransactionChecker transactionChecker;

    public SendMessageServiceImpl(String accessKey, String accessSecret, String endpoint, List<String> topics,
        TransactionChecker transactionChecker) {
        this.accessKey = accessKey;
        this.accessSecret = accessSecret;
        this.endpoint = endpoint;
        this.topics = topics;
        this.transactionChecker = transactionChecker;
    }

    @PostConstruct
    public void init() {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey,
            accessSecret);
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoint)
            .setCredentialProvider(sessionCredentialsProvider).build();
        ProducerBuilder builder = provider.newProducerBuilder().setClientConfiguration(clientConfiguration).setTopics(
            topics.toArray(new String[0]));
        if (transactionChecker != null) {
            builder.setTransactionChecker(transactionChecker);
        }
        try {
            producer = builder.build();
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 发送异步消息
     *
     * @param topic
     * @param tag
     * @param messageKey 消息键
     * @param body       身体
     */
    @Override
    public void sendAsyncMessage(String topic, String tag, String messageKey, String body) {
        Message message = buildMessage(topic, tag, messageKey, body);
        CompletableFuture<SendReceipt> future = producer.sendAsync(message);
        future.whenCompleteAsync((sendReceipt, throwable) -> {
            try {
                if (null != throwable) {
                    log.error("Send message failed, topic={} tag={} messageKey={} body={}", topic, tag, messageKey,
                        body, throwable);
                    return;
                }
                log.info("Send message successfully, topic={} tag={} messageId={} messageKey={} body={}", topic, tag,
                    sendReceipt.getMessageId(), messageKey, body);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }, sendCallbackExecutor);
    }

    @Override
    public <R> R sendTransactionMessage(String topic, String tag, String messageKey, String body,
        Function<String, R> consumer) throws Exception {
        if (transactionChecker == null) {
            throw new RuntimeException("TransactionChecker is null, not support send transaction message");
        }
        // 开启事务
        Transaction transaction = null;
        SendReceipt sendReceipt = null;
        R result = null;
        try {
            transaction = producer.beginTransaction();
            // 构建消息对象
            Message message = buildMessage(topic, tag, messageKey, body);
            // 发送半(half)事务消息
            sendReceipt = producer.send(message, transaction);
            log.info("Send transaction half message successfully, messageId={} messageKey={} topic={} tag={}",
                sendReceipt.getMessageId(), messageKey, topic, tag);
            // 执行业务逻辑
            result = consumer.apply(sendReceipt.getMessageId().toString());
        } catch (Throwable t) {
            // 失败则回滚事务消息
            rollback(topic, tag, messageKey, body, t, transaction, sendReceipt);
            throw t;
        }
        // 成功则提交事务消息
        commit(topic, tag, messageKey, transaction, sendReceipt);
        return result;
    }

    /**
     * 提交
     *
     * @param topic       主题
     * @param tag         标记
     * @param messageKey  消息键
     * @param transaction 交易
     * @param sendReceipt 发送回执
     */
    private void commit(String topic, String tag, String messageKey, Transaction transaction, SendReceipt sendReceipt) {
        try {
            transaction.commit();
            log.info("Commit transaction message successfully, messageId={} messageKey={} topic={} tag={}",
                sendReceipt.getMessageId(), messageKey, topic, tag);
        } catch (Throwable e) {
            // 提交失败不能影响业务执行，直接记录日志；rocketmq会调用 transactionChecker 来确认事务消息提交或回滚
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 回滚
     *
     * @param topic       主题
     * @param tag         标记
     * @param messageKey  消息键
     * @param body        身体
     * @param t           t
     * @param transaction 交易
     * @param sendReceipt
     */
    private void rollback(String topic, String tag, String messageKey, String body, Throwable t,
        Transaction transaction, SendReceipt sendReceipt) {
        log.error("Failed to send message or execute local transaction, messageKey={} topic={}"
            + " tag={} body={} message={}", messageKey, topic, tag, body, t.getMessage(), t);
        try {
            transaction.rollback();
            log.info("Rollback transaction message successfully, messageId={} messageKey={} topic={} tag={}",
                sendReceipt.getMessageId(), messageKey, topic, tag);
        } catch (Throwable e) {
            // 提交失败不能影响业务执行，直接记录日志；rocketmq会调用 transactionChecker 来确认事务消息提交或回滚
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 构建消息
     *
     * @param topic      主题
     * @param tag        标记
     * @param messageKey 消息键
     * @param body       身体
     * @return {@link Message }
     */
    private static Message buildMessage(String topic, String tag, String messageKey, String body) {
        MessageBuilder messageBuilder = provider.newMessageBuilder().setTopic(topic).setKeys(messageKey).setBody(
            body.getBytes(StandardCharsets.UTF_8));
        if (StringUtils.isNotBlank(tag)) {
            messageBuilder.setTag(tag);
        }
        return messageBuilder.build();
    }
}